package com.roy.netty.client;

import java.io.IOException;
import java.net.InetSocketAddress;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

public abstract class AbstractTCPClient implements NettyTCPClient {

	private Bootstrap bootstrap;
	private ClientSettings settings = new ClientSettings();
	private ServerLoadBalancer serverLoadBalancer;
	private NettyClientChannelInitializer channelInitializer;

	public AbstractTCPClient() {
		System.out.println("初始化Netty bootstrap");
		settings.setTransportServers("127.0.0.1:8089");
		 this.serverLoadBalancer = new RandomServerLoadBalancer(settings.getTransportServers());
		 this.channelInitializer = new NettyClientChannelInitializer(settings);
		EventLoopGroup bossGroup = new NioEventLoopGroup(
				settings.getTransportBossThreads());
		Class<? extends Channel> channelClazz = NioSocketChannel.class;
		if (settings.isTransportNativeEpoll()) {
			channelClazz = EpollSocketChannel.class;
		}
		this.bootstrap = new Bootstrap();
		this.bootstrap.group(bossGroup).channel(channelClazz)
				.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,
						settings.getTransportConnectTimeoutMillis())
				.option(ChannelOption.SO_REUSEADDR,
						settings.isTransportSoReuseAddress())
				.option(ChannelOption.TCP_NODELAY,
						settings.isTransportTcpNoDelay())
				.option(ChannelOption.SO_KEEPALIVE,
						settings.isTransportSoKeepAlive())
				.option(ChannelOption.SO_SNDBUF,
						settings.getTransportSoSendBufferSize())
				.option(ChannelOption.SO_RCVBUF,
						settings.getTransportSoReceiveBufferSize())
				.option(ChannelOption.SO_REUSEADDR,
						settings.isTransportSoReuseAddress());
	}
	@Override
	public void close() throws IOException {
		System.out.println("Closing NettyTCPClient...");
        bootstrap.group().shutdownGracefully().syncUninterruptibly();
        System.out.println("NettyTCPClient Closed.");
	}

	@Override
	public String send(String message, int waitMilliSeconds) throws Throwable {
		return sendToServer(message, waitMilliSeconds);
	}

	private String sendToServer(String message, int waitMilliSeconds) throws Throwable {
		InetSocketAddress address = this.serverLoadBalancer.getAddress();
		System.out.println("Send to server " + address.toString()
				+ " with message " + message + ", waiting for "
				+ String.valueOf(waitMilliSeconds) + " milliseconds");
		long start = System.currentTimeMillis();
		String responseMessage;
		try {
			responseMessage = doSend(address, message, waitMilliSeconds);
		} finally {
			//统计请求状态
//			clientStats.update(address.toString(),
//					System.currentTimeMillis() - start);
			System.out.println("request message "+message+" responsed in "+(System.currentTimeMillis() - start)+" milliSeconds");
		}
		System.out.println("Receive from server " + address + " with message "
				+ responseMessage);
		if (responseMessage == null) {
			throw new IllegalStateException(
					"Response of message " + message + " is null.");
		}
		return responseMessage;
	}
	public abstract String doSend(final InetSocketAddress address, final String message,
            final int waitMilliSeconds) throws Throwable;
	
	@Override
	public String send(String message) throws Throwable {
		return send(message, settings.getTransportFetchWaitMilliSeconds());

	}
	public Bootstrap getBootstrap() {
		return bootstrap;
	}
	
    protected MessageSender getMessageSender(Channel channel) {
        return (MessageSender) channel.pipeline().get("clientHandler");
    }
    
    protected NettyClientChannelInitializer getChannelInitializer() {
        return channelInitializer;
    }
    
    protected ClientSettings getSettings() {
		return this.settings;
	}

}
